package com.dtp.core.notify.manager;

import com.dtp.common.ApplicationContextHolder;
import com.dtp.common.em.NotifyItemEnum;
import com.dtp.common.entity.DtpExecutorProps;
import com.dtp.common.entity.NotifyItem;
import com.dtp.common.entity.NotifyPlatform;
import com.dtp.common.entity.TpExecutorProps;
import com.dtp.common.properties.DtpProperties;
import com.dtp.common.util.StreamUtil;
import com.dtp.core.support.ExecutorWrapper;
import com.dtp.core.thread.DtpExecutor;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.commons.collections.CollectionUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

import static com.dtp.common.em.NotifyItemEnum.CAPACITY;
import static com.dtp.common.em.NotifyItemEnum.LIVENESS;
import static com.dtp.common.em.NotifyItemEnum.QUEUE_TIMEOUT;
import static com.dtp.common.em.NotifyItemEnum.REJECT;
import static com.dtp.common.em.NotifyItemEnum.RUN_TIMEOUT;
import static com.dtp.common.entity.NotifyItem.mergeAllNotifyItems;
import static com.dtp.common.entity.NotifyItem.mergeSimpleNotifyItems;

/**
 * 通知帮助类
 *
 * @author yanhom
 * @since 1.0.0
 *
 * @remark muse
 */
@Slf4j
public class NotifyHelper {
    private static final List<String> COMMON_ALARM_KEYS = Lists.newArrayList("alarmType", "threshold");
    private static final Set<String> LIVENESS_ALARM_KEYS = Sets.newHashSet("corePoolSize", "maximumPoolSize", "poolSize", "activeCount");
    private static final Set<String> CAPACITY_ALARM_KEYS = Sets.newHashSet("queueType", "queueCapacity", "queueSize", "queueRemaining");
    private static final Set<String> REJECT_ALARM_KEYS = Sets.newHashSet("rejectType", "rejectCount");
    private static final Set<String> RUN_TIMEOUT_ALARM_KEYS = Sets.newHashSet("runTimeoutCount");
    private static final Set<String> QUEUE_TIMEOUT_ALARM_KEYS = Sets.newHashSet("queueTimeoutCount");

    private static final Set<String> ALL_ALARM_KEYS; // 存储所有ALARM_KEYS
    private static final Map<String, Set<String>> ALARM_KEYS = Maps.newHashMap(); // 维护NotifyItem与ALARM_KEYS

    static {
        // 维护ALARM_KEYS
        ALARM_KEYS.put(LIVENESS.name(), LIVENESS_ALARM_KEYS);
        ALARM_KEYS.put(CAPACITY.name(), CAPACITY_ALARM_KEYS);
        ALARM_KEYS.put(REJECT.name(), REJECT_ALARM_KEYS);
        ALARM_KEYS.put(RUN_TIMEOUT.name(), RUN_TIMEOUT_ALARM_KEYS);
        ALARM_KEYS.put(QUEUE_TIMEOUT.name(), QUEUE_TIMEOUT_ALARM_KEYS);

        // 维护ALL_ALARM_KEYS
        ALL_ALARM_KEYS = ALARM_KEYS.values().stream().flatMap(Collection::stream).collect(Collectors.toSet());
        ALL_ALARM_KEYS.addAll(COMMON_ALARM_KEYS);
    }

    private NotifyHelper() {}

    /** 获得所有ALARM_KEYS */
    public static Set<String> getAllAlarmKeys() {return ALL_ALARM_KEYS;}

    /** 根据notifyItem获得ALARM_KEYS */
    public static Set<String> getAlarmKeys(NotifyItemEnum notifyItemEnum) {
        val keys = ALARM_KEYS.get(notifyItemEnum.name());
        keys.addAll(COMMON_ALARM_KEYS);
        return keys;
    }

    /** 获得线程池executor中配置的notifyType类型的通知配置项信息 */
    public static Optional<NotifyItem> getNotifyItem(ExecutorWrapper executor, NotifyItemEnum notifyType) {
        return executor.getNotifyItems().stream()
                .filter(x -> notifyType.getValue().equalsIgnoreCase(x.getType())) // 只获取匹配notifyType的通知项
                .findFirst();
    }

    /** 向notifyItems中填充platformIds */
    @SuppressWarnings("unchecked")
    public static void fillPlatforms(List<String> platformIds, List<NotifyPlatform> platforms, List<NotifyItem> notifyItems) {
        // 要求platforms和notifyItems都存在
        if (CollectionUtils.isEmpty(platforms) || CollectionUtils.isEmpty(notifyItems)) {
            log.warn("DynamicTp notify, no notify platforms or items configured.");
            return;
        }

        // 从platforms集合中获得platformId集合
        List<String> globalPlatformIds = StreamUtil.fetchProperty(platforms, NotifyPlatform::getPlatformId);

        /** 为notifyItems赋值globalPlatformIds，优先级：notifyItems > platformIds > platforms（globalPlatformIds） */
        notifyItems.forEach(n -> {
            // 如果notifyItems中的NotifyItem元素有platformIds，则取platformIds和globalPlatformIds的交集赋值给NotifyItem元素
            if (CollectionUtils.isNotEmpty(n.getPlatformIds()))
                n.setPlatformIds((List<String>) CollectionUtils.intersection(globalPlatformIds, n.getPlatformIds()));

            // 如果notifyItems中没有platformIds，则判断如果入参platformIds不为空，则取platformIds和globalPlatformIds的交集
            else if (CollectionUtils.isNotEmpty(platformIds))
                n.setPlatformIds((List<String>) CollectionUtils.intersection(globalPlatformIds, platformIds));

            // 如果notifyItems中没有platformIds并且入参platformIds为空，需要兼容之前不存在platformIds的情况或取global
            else {
                if (CollectionUtils.isNotEmpty(n.getPlatforms())) // 如果NotifyItem中存储了平台名称（platform）
                    setPlatformIds(platforms, n);
                else
                    n.setPlatformIds(globalPlatformIds); // 否则设置platforms（globalPlatformIds）
            }
        });
    }
    /** 通过notifyItem中存储的平台名称（platform），去platforms中查找所有名称相同的平台，并获取platformId集合，维护到notifyItem中 */
    private static void setPlatformIds(List<NotifyPlatform> platforms, NotifyItem notifyItem) {
        List<String> platformIds = new ArrayList<>();
        // 维护映射：key=平台名称（platform） value=NotifyPlatform实例对象
        Map<String, NotifyPlatform> platformMap = StreamUtil.toMap(platforms, NotifyPlatform::getPlatform);

        // 通过入参notifyItem中存储的平台名称（platform）去映射platformMap中查找NotifyPlatform实例对象，并且将该实例中的platformId维护到platformIds中
        for (String platform : notifyItem.getPlatforms()) {
            NotifyPlatform notifyPlatform = platformMap.get(platform);
            if (notifyPlatform != null)
                platformIds.add(notifyPlatform.getPlatformId());
        }
        notifyItem.setPlatformIds(platformIds);
    }

    /** 通过platformId获得NotifyPlatform实例对象 */
    public static Optional<NotifyPlatform> getPlatform(String platformId) {
        Map<String, NotifyPlatform> platformMap = getAllPlatforms();
        return Optional.ofNullable(platformMap.get(platformId));
    }
    /** 从dtp的配置文件信息中获得平台映射信息，即：key=platformId, value=NotifyPlatform实例对象 */
    public static Map<String, NotifyPlatform> getAllPlatforms() {
        // 获得dtp的配置信息，
        val dtpProperties = ApplicationContextHolder.getBean(DtpProperties.class);
        if (Objects.isNull(dtpProperties) || CollectionUtils.isEmpty(dtpProperties.getPlatforms()))
            return Collections.emptyMap();

        // 尝试从配置项"spring.dynamic.tp.platforms"下的NotifyPlatform中获得platformId列表
        return StreamUtil.toMap(dtpProperties.getPlatforms(), NotifyPlatform::getPlatformId);
    }

    /** 初始化通知操作 */
    public static void initNotify(DtpExecutor executor) {
        // 获得最新的dtp配置信息，如果为空，则直接返回
        val dtpProperties = ApplicationContextHolder.getBean(DtpProperties.class);
        if (Objects.isNull(dtpProperties)) return;

        // 获得配置的"通知报警平台配置"（spring.dynamic.tp.platforms），如果为空，则直接返回
        val platforms = dtpProperties.getPlatforms();
        if (CollectionUtils.isEmpty(platforms)) {
            executor.setNotifyItems(Lists.newArrayList());
            executor.setPlatformIds(Lists.newArrayList());
            log.warn("DynamicTp notify, no notify platforms configured for [{}]", executor.getThreadPoolName());
            return;
        }

        // 如果线程池中的通知项（spring.dynamic.tp.executors.notifyItems）就没有被配置，则直接返回
        if (CollectionUtils.isEmpty(executor.getNotifyItems())) return;

        // 向notifyItems中填充platformIds
        fillPlatforms(executor.getPlatformIds(), platforms, executor.getNotifyItems());

        // 初始化报警管理器
        AlarmManager.initAlarm(executor.getThreadPoolName(), executor.getNotifyItems());
    }

    /** 更新ExecutorWrapper的通知项 */
    public static void updateNotifyInfo(ExecutorWrapper executorWrapper, TpExecutorProps props, List<NotifyPlatform> platforms) {
        // 将最新的dtp配置中的通知项与简单通知项（getSimpleNotifyItems）执行合并操作
        val allNotifyItems = mergeSimpleNotifyItems(props.getNotifyItems());

        // 执行刷新通知操作
        refreshNotify(executorWrapper.getThreadPoolName(),
                props.getPlatformIds(),
                platforms,
                executorWrapper.getNotifyItems(),
                allNotifyItems);
        executorWrapper.setNotifyItems(allNotifyItems);
        executorWrapper.setPlatformIds(props.getPlatformIds());
        executorWrapper.setNotifyEnabled(props.isNotifyEnabled());
    }

    /** 更新DtpExecutor的通知项 */
    public static void updateNotifyInfo(DtpExecutor executor, DtpExecutorProps props, List<NotifyPlatform> platforms) {
        // 将最新的dtp配置中的通知项与全部通知项（getAllNotifyItems）执行合并操作
        val allNotifyItems = mergeAllNotifyItems(props.getNotifyItems());

        // 执行刷新通知操作
        refreshNotify(executor.getThreadPoolName(),
                props.getPlatformIds(),
                platforms,
                executor.getNotifyItems(), // oldNotifyItems
                allNotifyItems); // newNotifyItems
        executor.setNotifyItems(allNotifyItems);
        executor.setPlatformIds(props.getPlatformIds());
        executor.setNotifyEnabled(props.isNotifyEnabled());
    }

    /** 执行刷新通知操作 */
    private static void refreshNotify(String poolName, List<String> platformIds, List<NotifyPlatform> platforms,
                                      List<NotifyItem> oldNotifyItems, List<NotifyItem> newNotifyItems) {
        // 向notifyItems中填充platformIds
        fillPlatforms(platformIds, platforms, newNotifyItems);

        // 遍历新的通知项，只有当旧的通知项中配置的interval值与新的通知项中配置的interval值不同时，才会执行initAlarm操作
        Map<String, NotifyItem> oldNotifyItemMap = StreamUtil.toMap(oldNotifyItems, NotifyItem::getType);
        newNotifyItems.forEach(x -> {
            NotifyItem oldNotifyItem = oldNotifyItemMap.get(x.getType());
            // 跳出本元素，遍历下个元素
            if (Objects.nonNull(oldNotifyItem) && oldNotifyItem.getInterval() == x.getInterval()) return;

            // 初始化报警管理器
            AlarmManager.initAlarm(poolName, x);
        });
    }
}
